package com.microframework.base.core.redis.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.stereotype.Service;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Redis消息发布/订阅服务
 */
@Slf4j
@Service
public class RedisPubSubService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private RedisMessageListenerContainer redisMessageListenerContainer;
    
    // 存储订阅信息，方便取消订阅
    private final ConcurrentHashMap<String, List<Topic>> topicMap = new ConcurrentHashMap<>();

    /**
     * 发布消息
     *
     * @param channel 频道
     * @param message 消息内容
     */
    public void publish(String channel, Object message) {
        try {
            redisTemplate.convertAndSend(channel, message);
            log.debug("消息发布成功 - 频道: {}, 消息: {}", channel, message);
        } catch (Exception e) {
            log.error("消息发布失败 - 频道: {}, 消息: {}", channel, message, e);
            throw e;
        }
    }

    /**
     * 订阅频道
     *
     * @param channel 频道
     * @param listener 消息监听器
     */
    public void subscribe(String channel, MessageListener listener) {
        try {
            ChannelTopic topic = new ChannelTopic(channel);
            redisMessageListenerContainer.addMessageListener(listener, topic);
            
            // 记录订阅信息
            topicMap.computeIfAbsent(channel, k -> new ArrayList<>()).add(topic);
            log.debug("频道订阅成功: {}", channel);
        } catch (Exception e) {
            log.error("频道订阅失败: {}", channel, e);
            throw e;
        }
    }
    
    /**
     * 使用模式订阅频道
     *
     * @param pattern 模式，如"channel.*"
     * @param listener 消息监听器
     */
    public void subscribePattern(String pattern, MessageListener listener) {
        try {
            PatternTopic topic = new PatternTopic(pattern);
            redisMessageListenerContainer.addMessageListener(listener, topic);
            
            // 记录订阅信息
            topicMap.computeIfAbsent(pattern, k -> new ArrayList<>()).add(topic);
            log.debug("模式频道订阅成功: {}", pattern);
        } catch (Exception e) {
            log.error("模式频道订阅失败: {}", pattern, e);
            throw e;
        }
    }
    
    /**
     * 取消订阅
     *
     * @param channel 频道
     * @param listener 消息监听器
     */
    public void unsubscribe(String channel, MessageListener listener) {
        try {
            List<Topic> topics = topicMap.get(channel);
            if (topics != null) {
                topics.forEach(topic -> redisMessageListenerContainer.removeMessageListener(listener, topic));
                topicMap.remove(channel);
                log.debug("取消订阅成功: {}", channel);
            }
        } catch (Exception e) {
            log.error("取消订阅失败: {}", channel, e);
            throw e;
        }
    }
} 